package com.shujia.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo21PageRank {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()

    conf.setAppName("acc")

    conf.setMaster("local")

    val sc = new SparkContext(conf)

    //1、读取网页数据
    val linesRDD: RDD[String] = sc.textFile("data/pagerank.txt")

    //2、整理数据
    val kvRDD: RDD[(String, List[String])] = linesRDD.map((line: String) => {
      val split: Array[String] = line.split("-")
      //当前网页
      val page: String = split(0)
      //出链列表
      val linkedList: List[String] = split(1).split(",").toList
      (page, linkedList)
    })

    //计算网页数量
    val N: Long = kvRDD.count()

    //3、给每一个网页一个初始的pr值
    var pageRankRDD: RDD[(String, List[String], Double)] = kvRDD.map {
      case (page: String, linkedList: List[String]) => (page, linkedList, 1.0)
    }

    //收敛标准
    val p: Double = 0.0001
    var flag = true
    //阻尼系数
    val q = 0.85

    while (flag) {
      //4、将每个网页 的pr值平分给出链列表
      val agePrRDD: RDD[(String, Double)] = pageRankRDD.flatMap {
        case (page: String, linkedList: List[String], pr: Double) =>
          //计算平分pr值
          val avgPr: Double = pr / linkedList.length
          //在出链列表后面增加分到的pr值
          linkedList.map((pg: String) => (pg, avgPr))
      }
      //5、对分到的pr求和
      val newPrRDD: RDD[(String, Double)] = agePrRDD
        .reduceByKey(_ + _)
        //增加阻尼系数
        .map {
          case (page: String, pr: Double) => {
            (page, (1 - q) / N + q * pr)
          }
        }

      //6、关联获取到原始的出链列表
      val joinRDD: RDD[(String, (Double, List[String]))] = newPrRDD.join(kvRDD)
      //7、整理格式
      val newPageRankRDD: RDD[(String, List[String], Double)] = joinRDD.map {
        case (page: String, (pr: Double, linkedList: List[String])) => (page, linkedList, pr)
      }

      //8、计算新的pr和旧的pr的差值的平均值
      val prczRDD: RDD[Double] = pageRankRDD
        .map {
          case (page: String, linkedList: List[String], pr: Double) => (page, pr)
        }
        .join(newPrRDD) //关联新的pr值
        .map {
          case (page: String, (pr: Double, newPr: Double)) => Math.abs(pr - newPr)
        }

      //差值平均值
      val chaAvgPr: Double = prczRDD.sum() / prczRDD.count()

      println(s"差值平均值：$chaAvgPr")

      //收敛
      if (chaAvgPr < p) {
        flag = false
      }


      //切换rdd,供下一次迭代使用
      pageRankRDD = newPageRankRDD
      newPageRankRDD.foreach(println)
    }


    pageRankRDD.foreach(println)
  }


}
